package org.cttic.visualization.visualization.service.impl;

import cn.hutool.json.JSONUtil;
import com.alibaba.fastjson.JSONObject;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.cttic.visualization.visualization.common.commonConf;
import org.cttic.visualization.visualization.common.realTimeData;
import org.cttic.visualization.visualization.util.DataUtil;
import org.cttic.visualization.visualization.util.ParseJsonDataUtil;
import org.cttic.visualization.visualization.websocket.Data;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaHandler;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import javax.lang.model.element.VariableElement;
import java.text.ParseException;
import java.util.ArrayList;
import java.util.List;


/**
 * @author shkstart
 * @create 2020-04-28 18:14
 * 2.2.3实时推送途经识别点接口
 */
@Component
@Slf4j
public class realTimeImpl {

    List<String> list;
    @Autowired
    DataUtil dataUtil;

    //指定监听的topic，当前消费者组id
    //@KafkaHandler
    //@KafkaListener(topics = commonConf.kafkaTopic)
    public void registryReceiver(List<ConsumerRecord<Integer, String>> records) throws ParseException {


        if (records.size() >= 10) {
//            for(int i=0;i<5;i++) {

            String value = records.get(0).value();
            realTimeData data = dataUtil.sendDataUtil(value);
            Data.sendMessage(JSONUtil.parseObj(data));
//            }
        } else {
            for (int i = 0; i < records.size(); i++) {
                double v = Math.random() * 10;
                if (v >= 9) {
                    realTimeData data = dataUtil.sendDataUtil(records.get(0).value());
                    ObjectMapper mapper = new ObjectMapper();
                    try {
                        String json = mapper.writeValueAsString(data);
                    } catch (JsonProcessingException e) {
                        e.printStackTrace();
                    }
                    Data.sendMessage(JSONUtil.parseObj(data));
                }

            }
        }
//        list.add(records.value());
//        if(list.size()>=100){
//            realTimeData data = dataUtil.sendDataUtil(list.get(0));
//            System.out.println("发送数据："+list.get(0));
//            Data.sendMessage(JSONUtil.parseObj(data));
//            list.clear();
//        }

    }

}